package com.example;

import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.serialization.StringSerializer;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ProducerDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException, NoSuchFieldException, IllegalAccessException {

        KafkaProducer<String,String> producer = new KafkaProducer<>(initConfig());


        int index = 0;
        while (true){
            ProducerRecord<String,String> record = new ProducerRecord("ACKS_0","1");
            Future<RecordMetadata> future = producer.send(record);
            RecordMetadata recordMetadata = future.get();
            long offset = recordMetadata.offset();
            int partition = recordMetadata.partition();
            System.out.println("index=" + index + " 已发送,partition="+ partition+" offset="+offset);
            index++;
            Thread.sleep(4000);
        }






    }


    private static Properties initConfig(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.35.101:9094");
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.199.200:9094,192.168.199.200:9095,192.168.199.200:9096");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"15");
        properties.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG,"2");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
//        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,InterceptorDemo.class.getName());

        return properties;
    }
}
